package belf.migrate.api.queue;

import org.redisson.api.RQueue;
import org.redisson.api.RedissonClient;

public class RedisQueue {
    public static final String QUEUE_NAME = "belf_queue";
    private static final RedissonClient redissonClient = RedisClient.getInstance().getConnection();

    /**
     * 从debezium获得的增量数据放入redis队列
     * @param jsonMessage 从debezium获得的原始的增量事件ChangedEvent是一个json消息
     */
    public static void push(String jobId, String jsonMessage) {
        RQueue<String> queue = redissonClient.getQueue(QUEUE_NAME + "_" + jobId);
        queue.add(jsonMessage);
    }

    /**
     * 从redis队列中获取一个json消息，准备插入目标数据库
     * @return
     */
    public static String pop(String jobId) {
        RQueue<String> queue = redissonClient.getQueue(QUEUE_NAME + "_" + jobId);
        return queue.poll();
    }

    public static String peek(String jobId) {
        RQueue<String> queue = redissonClient.getQueue(QUEUE_NAME + "_" + jobId);
        return queue.peek();
    }

    public static boolean isEmpty(String jobId) {
        RQueue<String> queue = redissonClient.getQueue(QUEUE_NAME + "_" + jobId);
        return queue.isEmpty();
    }
}
